package com.zcccccy.rocketmq.consumer.config;

import com.zcccccy.rocketmq.consumer.config.properties.ConsumerProperties;
import com.zcccccy.rocketmq.consumer.listener.RocketMQListener;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * 消费者创建
 * 
 * @projectName: zcccccy
 * @packageName: com.zcccccy.extend.rocketmq.consumer.config
 * @author: zcy
 * @createTime: 2024/1/27
 * @productName: For Mac IntelliJ IDEA
 */

@Configuration
@ConditionalOnProperty(value = "rocketmq.consumer.enabled", havingValue = "true")
public class ConsumerConfig {

    private static final Logger log = LoggerFactory.getLogger(ConsumerConfig.class);

    @Autowired
    private ConsumerProperties consumerProperties;

    @Bean
    public void builderPushConsumer() {
        // TODO 本项目消费方式使用PushConsumer消费，如需要SimpleConsumer消费机制请参照官网示例

        log.info("==============》 开始初始化消费者 《===================");

        // TODO 每个主题对应一种消息类型，所以初始化消费多个主题以达到消费多种类型消息。本项目只对接了普通消息和定时/延时消息。
        // 其余消息可参照官网：https://rocketmq.apache.org/zh/docs/domainModel/04message

        // 主题构建
        Map<String, FilterExpression> topicMap = new HashMap<>();
        for (String topic : consumerProperties.getTopicList()) {
            // 订阅消息的过滤规则，表示订阅所有Tag的消息。
            String tag = "*";
            FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

            topicMap.put(topic, filterExpression);
        }

        try {

            final ClientServiceProvider provider = ClientServiceProvider.loadService();

            // 连接信息构建
            ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                // 端点
                .setEndpoints(consumerProperties.getEndpoints())
                .build();

            // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
            PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                //
                .setClientConfiguration(clientConfiguration)
                // 设置消费者分组。
                .setConsumerGroup(consumerProperties.getGroup())
                // 设置预绑定的订阅关系。
                .setSubscriptionExpressions(topicMap)
                // 设置消费监听器。
                .setMessageListener(new RocketMQListener())
                .build();
            // 如果不需要再使用PushConsumer，可关闭该进程。
            // pushConsumer.close();

            log.info("==============》 初始化消费者成功：{}", pushConsumer);

        } catch (ClientException e) {

            log.error("==============》 初始化消费者失败：{}", e.getMessage());
        }

    }

}
